package com.leon.datalink.core.persistence;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import com.leon.datalink.core.persistence.event.*;
import com.leon.datalink.core.persistence.state.NodeState;
import com.leon.datalink.core.persistence.state.NodeStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 第五阶段端到端集成测试
 * 
 * 验证DataLink Actor持久化功能的完整流程：
 * 1. 配置管理端到端测试
 * 2. Actor创建和持久化测试
 * 3. 状态恢复和一致性测试
 * 4. UI配置集成测试
 * 5. 性能监控集成测试
 * 6. 错误处理和边界条件测试
 * 
 * @author leon
 */
public class Phase5EndToEndIntegrationTest {
    
    private static final Logger logger = LoggerFactory.getLogger(Phase5EndToEndIntegrationTest.class);
    
    private ActorSystem actorSystem;
    private PersistenceConfigManager configManager;
    private SnapshotStrategyManager snapshotManager;
    private PersistenceMetricsManager metricsManager;
    
    public static void main(String[] args) {
        logger.info("=== DataLink Actor持久化功能第五阶段端到端集成测试 ===");
        
        try {
            Phase5EndToEndIntegrationTest test = new Phase5EndToEndIntegrationTest();
            test.runEndToEndIntegrationTest();
            
            logger.info("=== 第五阶段端到端集成测试完成 ===");
            
        } catch (Exception e) {
            logger.error("第五阶段端到端集成测试失败", e);
            System.exit(1);
        }
    }
    
    public void runEndToEndIntegrationTest() throws Exception {
        logger.info("开始第五阶段端到端集成测试...");
        
        try {
            // 1. 初始化测试环境
            initializeTestEnvironment();
            
            // 2. 配置管理端到端测试
            testConfigurationManagement();
            
            // 3. Actor生命周期端到端测试
            testActorLifecycleEndToEnd();
            
            // 4. 持久化和恢复端到端测试
            testPersistenceAndRecoveryEndToEnd();
            
            // 5. 智能快照策略端到端测试
            testSnapshotStrategyEndToEnd();
            
            // 6. 性能监控端到端测试
            testPerformanceMonitoringEndToEnd();
            
            // 7. 错误处理和边界条件测试
            testErrorHandlingAndBoundaryConditions();
            
            // 8. 多Actor并发测试
            testMultiActorConcurrency();
            
            logger.info("第五阶段端到端集成测试全部通过！");
            
        } finally {
            // 清理测试环境
            cleanupTestEnvironment();
        }
    }
    
    /**
     * 初始化测试环境
     */
    private void initializeTestEnvironment() throws Exception {
        logger.info("初始化测试环境...");
        
        // 创建ActorSystem
        actorSystem = ActorSystem.create("datalink-test");
        
        // 初始化管理器
        configManager = PersistenceConfigManager.getInstance();
        snapshotManager = SnapshotStrategyManager.getInstance();
        metricsManager = PersistenceMetricsManager.getInstance();
        
        // 清空之前的数据
        metricsManager.clearMetrics();
        
        // 设置测试配置
        configManager.setGlobalPersistenceEnabled(true);
        configManager.setActorTypePersistence("TestActor", true);
        
        logger.info("✓ 测试环境初始化完成");
    }
    
    /**
     * 配置管理端到端测试
     */
    private void testConfigurationManagement() {
        logger.info("测试配置管理端到端流程...");
        
        // 测试全局配置
        configManager.setGlobalPersistenceEnabled(false);
        assert !configManager.isGlobalPersistenceEnabled() : "全局配置设置失败";
        
        configManager.setGlobalPersistenceEnabled(true);
        assert configManager.isGlobalPersistenceEnabled() : "全局配置恢复失败";
        
        // 测试Actor类型配置
        String actorType = "com.leon.datalink.rule.actor.TestActor";
        configManager.setActorTypePersistence(actorType, true);

        Map<String, Object> testProperties = new HashMap<>();
        testProperties.put("test", "value");
        assert configManager.shouldPersist("test-node", actorType, testProperties) : "Actor类型配置失败";

        // 测试实例级配置
        String nodeId = "test-node-001";
        configManager.setInstancePersistence(nodeId, false);
        assert !configManager.shouldPersist(nodeId, actorType, testProperties) : "实例级配置失败";

        // 测试消息过滤（简化测试）
        assert configManager.shouldPersistMessage(nodeId, "NormalMessage") : "正常消息应该可以持久化";
        
        logger.info("✓ 配置管理端到端测试通过");
    }
    
    /**
     * Actor生命周期端到端测试（简化版本）
     */
    private void testActorLifecycleEndToEnd() throws Exception {
        logger.info("测试Actor生命周期端到端流程...");

        String actorType = "TestActor";
        String nodeId = "lifecycle-test-node";

        // 确保配置启用
        configManager.setActorTypePersistence(actorType, true);

        // 模拟Actor生命周期事件
        Map<String, Object> properties = new HashMap<>();
        properties.put("test.property", "test.value");
        properties.put("lifecycle.test", true);

        // 验证配置生效
        assert configManager.shouldPersist(nodeId, actorType, properties) : "Actor配置未生效";

        // 模拟持久化操作
        for (int i = 0; i < 10; i++) {
            metricsManager.recordPersistOperation(actorType, nodeId, 20 + i, 1024 + i * 100);
            Thread.sleep(50);
        }

        // 验证性能指标
        PersistenceMetricsManager.ActorMetrics metrics = metricsManager.getActorMetrics(actorType, nodeId);
        assert metrics != null : "性能指标未记录";
        assert metrics.getTotalPersistCount() == 10 : "持久化计数不正确";

        logger.info("✓ Actor生命周期端到端测试通过");
    }
    
    /**
     * 持久化和恢复端到端测试（简化版本）
     */
    private void testPersistenceAndRecoveryEndToEnd() throws Exception {
        logger.info("测试持久化和恢复端到端流程...");

        String actorType = "PersistenceTestActor";
        String nodeId = "persistence-recovery-node";

        // 第一阶段：模拟持久化操作
        Map<String, Object> properties = new HashMap<>();
        properties.put("recovery.test", "phase1");

        // 模拟多个持久化事件
        for (int i = 0; i < 20; i++) {
            metricsManager.recordPersistOperation(actorType, nodeId, 30 + i, 1024 + i * 50);
            Thread.sleep(20);
        }

        // 记录第一阶段的指标
        PersistenceMetricsManager.ActorMetrics metrics1 = metricsManager.getActorMetrics(actorType, nodeId);
        long persistCount1 = metrics1.getTotalPersistCount();

        // 第二阶段：模拟恢复操作
        for (int i = 0; i < 5; i++) {
            metricsManager.recordRecoveryOperation(actorType, nodeId, 100 + i * 10, 4);
            Thread.sleep(50);
        }

        // 继续模拟更多持久化操作
        for (int i = 20; i < 30; i++) {
            metricsManager.recordPersistOperation(actorType, nodeId, 30 + i, 1024 + i * 50);
            Thread.sleep(20);
        }

        // 验证恢复后的指标
        PersistenceMetricsManager.ActorMetrics metrics2 = metricsManager.getActorMetrics(actorType, nodeId);
        assert metrics2.getTotalPersistCount() > persistCount1 : "恢复后持久化计数未增加";
        assert metrics2.getTotalRecoveryCount() > 0 : "恢复计数为0";

        logger.info("✓ 持久化和恢复端到端测试通过");
    }
    
    /**
     * 智能快照策略端到端测试
     */
    private void testSnapshotStrategyEndToEnd() throws Exception {
        logger.info("测试智能快照策略端到端流程...");
        
        String actorType = "com.leon.datalink.rule.actor.RuleActor";
        String nodeId = "snapshot-strategy-node";
        
        // 获取快照策略
        SnapshotStrategyManager.SnapshotStrategy strategy = snapshotManager.getSnapshotStrategy(actorType, nodeId);
        assert strategy != null : "快照策略获取失败";
        
        int originalFrequency = strategy.getSnapshotFrequency();
        
        // 模拟高频状态变化
        NodeState oldState = createTestNodeState(nodeId);
        for (int i = 0; i < 50; i++) {
            NodeState newState = createTestNodeState(nodeId);
            newState.setProcessedCount(i);
            snapshotManager.updateStateChangeStats(actorType, nodeId, oldState, newState);
            oldState = newState;
            Thread.sleep(10);
        }
        
        // 验证策略是否自适应调整
        Thread.sleep(1000);
        
        // 测试快照判断逻辑
        NodeState testState = createTestNodeState(nodeId);
        boolean shouldSnapshot = snapshotManager.shouldCreateSnapshot(actorType, nodeId, 
                                                                     strategy.getSnapshotFrequency(), 
                                                                     testState);
        assert shouldSnapshot : "快照判断逻辑失败";
        
        // 记录快照创建
        snapshotManager.recordSnapshotCreated(actorType, nodeId, 2048);
        assert strategy.getSnapshotCount() > 0 : "快照计数未更新";
        
        logger.info("✓ 智能快照策略端到端测试通过");
    }
    
    /**
     * 性能监控端到端测试
     */
    private void testPerformanceMonitoringEndToEnd() throws Exception {
        logger.info("测试性能监控端到端流程...");
        
        String actorType = "PerformanceTestActor";
        String nodeId = "performance-monitor-node";
        
        // 模拟各种操作
        for (int i = 0; i < 100; i++) {
            // 模拟持久化操作
            long persistTime = 10 + (i % 50);
            long dataSize = 1024 + (i * 10);
            metricsManager.recordPersistOperation(actorType, nodeId, persistTime, dataSize);
            
            // 模拟恢复操作
            if (i % 10 == 0) {
                long recoveryTime = 50 + (i % 100);
                metricsManager.recordRecoveryOperation(actorType, nodeId, recoveryTime, 5);
            }
            
            // 模拟快照操作
            if (i % 20 == 0) {
                long snapshotTime = 100 + (i % 200);
                long snapshotSize = 2048 + (i * 20);
                metricsManager.recordSnapshotOperation(actorType, nodeId, snapshotTime, snapshotSize);
            }
        }
        
        // 验证Actor级别指标
        PersistenceMetricsManager.ActorMetrics actorMetrics = metricsManager.getActorMetrics(actorType, nodeId);
        assert actorMetrics != null : "Actor指标未记录";
        assert actorMetrics.getTotalPersistCount() == 100 : "持久化计数不正确";
        assert actorMetrics.getAvgPersistLatency() > 0 : "平均延迟计算错误";
        
        // 验证系统级别指标
        PersistenceMetricsManager.SystemMetrics systemMetrics = metricsManager.getSystemMetrics();
        assert systemMetrics.getTotalPersistCount() > 100 : "系统级别计数不正确";
        assert systemMetrics.getPersistThroughput() >= 0 : "吞吐量计算错误";
        
        // 生成性能报告
        String report = metricsManager.generatePerformanceReport();
        assert report != null && !report.isEmpty() : "性能报告生成失败";
        assert report.contains("系统级别指标") : "报告内容不完整";
        
        logger.info("性能监控报告:\n{}", report);
        logger.info("✓ 性能监控端到端测试通过");
    }
    
    /**
     * 错误处理和边界条件测试
     */
    private void testErrorHandlingAndBoundaryConditions() {
        logger.info("测试错误处理和边界条件...");
        
        // 测试空值处理
        try {
            configManager.shouldPersist(null, "test", new HashMap<>());
            assert false : "空值检查失败";
        } catch (Exception e) {
            // 预期的异常
        }
        
        // 测试无效配置
        configManager.setGlobalPersistenceEnabled(false);
        Map<String, Object> testProps = new HashMap<>();
        assert !configManager.shouldPersist("test", "TestActor", testProps) : "全局禁用时应该返回false";

        // 恢复配置
        configManager.setGlobalPersistenceEnabled(true);
        
        // 测试大量数据处理
        String actorType = "BoundaryTestActor";
        for (int i = 0; i < 1000; i++) {
            metricsManager.recordPersistOperation(actorType, "node" + i, 1, 1024);
        }
        
        // 验证系统稳定性
        PersistenceMetricsManager.SystemMetrics metrics = metricsManager.getSystemMetrics();
        assert metrics.getTotalPersistCount() > 1000 : "大量数据处理失败";
        
        logger.info("✓ 错误处理和边界条件测试通过");
    }
    
    /**
     * 多Actor并发测试
     */
    private void testMultiActorConcurrency() throws Exception {
        logger.info("测试多Actor并发场景...");
        
        CountDownLatch latch = new CountDownLatch(5);
        
        // 模拟多个Actor的并发操作
        for (int i = 0; i < 5; i++) {
            String nodeId = "concurrent-node-" + i;
            String actorType = "ConcurrentTestActor";

            // 并发模拟持久化操作
            CompletableFuture.runAsync(() -> {
                try {
                    for (int j = 0; j < 20; j++) {
                        metricsManager.recordPersistOperation(actorType, nodeId, 25 + j, 1024 + j * 30);
                        Thread.sleep(25);
                    }
                } catch (Exception e) {
                    logger.error("并发测试异常", e);
                } finally {
                    latch.countDown();
                }
            });
        }
        
        // 等待所有任务完成
        latch.await(30, TimeUnit.SECONDS);
        Thread.sleep(2000);
        
        // 验证并发安全性
        Map<String, PersistenceMetricsManager.ActorMetrics> allMetrics = metricsManager.getAllActorMetrics();
        long totalConcurrentPersist = allMetrics.entrySet().stream()
                .filter(entry -> entry.getKey().contains("ConcurrentTestActor"))
                .mapToLong(entry -> entry.getValue().getTotalPersistCount())
                .sum();
        
        assert totalConcurrentPersist >= 100 : "并发持久化计数不正确: " + totalConcurrentPersist;
        
        logger.info("✓ 多Actor并发测试通过");
    }
    
    /**
     * 清理测试环境
     */
    private void cleanupTestEnvironment() {
        logger.info("清理测试环境...");
        
        if (actorSystem != null) {
            actorSystem.terminate();
        }
        
        if (metricsManager != null) {
            metricsManager.shutdown();
        }
        
        logger.info("✓ 测试环境清理完成");
    }
    
    /**
     * 创建测试节点状态
     */
    private NodeState createTestNodeState(String nodeId) {
        NodeState state = new NodeState();
        state.setNodeId(nodeId);
        state.setStatus(NodeStatus.RUNNING);
        state.setProcessedCount(0);
        state.setLastEventTime(System.currentTimeMillis());
        
        Map<String, Object> properties = new HashMap<>();
        properties.put("test", "value");
        state.setProperties(properties);
        
        return state;
    }
    
    /**
     * 测试用的Mock Actor类
     */
    public static class TestMockActor {
        // 简单的Mock Actor，用于测试
    }
}
